package com.dada.cn.paimon;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class BatchRead {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.enableCheckpointing(60000);
        env.setStateBackend(new FsStateBackend("file:///opt/soft/checkpoint/paimon"));
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        TableConfig tableConfig = tEnv.getConfig();

        tableConfig.set("changelog-producer.row-deduplicate", "true");

        String catalog = "CREATE CATALOG paimon_file WITH (\n" +
                "  'type' = 'paimon',\n" +
                "  'warehouse' = 'file:///opt/soft/paimon'\n" +
                ");";
        String useCatalog = "USE CATALOG paimon_file;\n";

//        String rest = "call sys.reset_consumer(`table` => 'default.test', consumer_id => 'myid');";
//        String select = "select * from test/*+ OPTIONS('scan.snapshot-id' = '0') */;";
        /*+ OPTIONS('consumer-id' = 'myid', 'consumer.mode' = 'at-least-once') */
        String select = "select * from test/*+ OPTIONS('consumer-id' = 'myid3', 'scan.snapshot-id' = '2','consumer.expiration-time' = '1 d','consumer.mode' = 'exactly-once') */;";

        tEnv.executeSql(catalog);
        tEnv.executeSql(useCatalog);
//        tEnv.executeSql(rest);
        tEnv.executeSql(select).print();




    }

}
